#include "server.h"

typedef int (*_P_SVC_FUNC)(struct worker_ctl *, char *);
#define _SVC_FUNC(_svc)  \
static int _SVC_FUNC_##_svc(struct worker_ctl *worker_ctl, char *msg)

_SVC_FUNC(NEW_SLOT)
{/*{{{*/
    PARA para;
    char *value;
    SOCKET new_sock;
    EP_SLOT new_slot;
    int res;

    para_load(msg, "sock(number:6)", &para);
    para_ref_by_name(para, "sock", &value);
    new_sock=atoi(value);
    para_free(para);

    if((res=ep_slot_register(worker_ctl->ep, new_sock, IS_FALSE,
                    &new_slot)))
    {
        errlog("%s: worker(%d) ep_slot_register fail, res:%d",
                __func__, worker_ctl->worker_idx, res);
        sock_close(new_sock);
        return(-1);
    }
    ep_slot_set_wrap(new_slot);

    errlog("%s: worker(%d) new slot", __func__, worker_ctl->worker_idx);
    return(0);
}/*}}}*/

static int _send(int worker_idx, char *msg_in, char *msg_out,
        unsigned int is_vip)
{/*{{{*/
    PARA para;
    char *rcpt, *from, *reply, *subject, *body;
    char curr_dt[DATE_SIZE+1];
    unsigned int mail_id;
    struct mail *mail;
    int sender_idx;
    char mail_file[LONG_FILE_NM_SIZE+1];
    FILE *fp;
    int res;

    if((res=para_load(msg_in,
                    "rcpt(string:1-200)"
                    "from(string:1-80)"
                    "reply(string:1-80)"
                    "subject(string:1-200)"
                    "body(string:1-65535)",
                    &para)))
    {
        para_put_item(msg_out, SOAR_ERR_PARA);
        goto RESPOND;
    }
    para_ref_by_name(para, "rcpt", &rcpt);
    para_ref_by_name(para, "from", &from);
    para_ref_by_name(para, "reply", &reply);
    para_ref_by_name(para, "subject", &subject);
    para_ref_by_name(para, "body", &body);

    // 检查黑名单
    if(rb_find(&share_info->black_list, (void *)rcpt))
    {
        errlog("%s: %s in BLACK", __func__, rcpt);
        para_put_item(msg_out, SOAR_ERR_MAILFWD_BLACK);
        goto RESPOND;
    }

    // 组织给sender的消息
    curr_date(curr_dt);
    pthread_mutex_lock(&share_info->mail_id_lock);
    if(strcmp(share_info->curr_dt, curr_dt))
    {
        share_info->mail_id=1;
        strcpy(share_info->curr_dt, curr_dt);
    }
    mail_id=share_info->mail_id;
    share_info->mail_id++;
    pthread_mutex_unlock(&share_info->mail_id_lock);

    ALLOC_DOERR(mail, struct mail,
            para_put_item(msg_out, SOAR_ERR_RESOURCE); goto RESPOND);
    mail->try_times=0;
    if(is_vip)
        sprintf(mail->file, "%s_%09u_vip.txt", curr_dt, mail_id);
    else
        sprintf(mail->file, "%s_%09u.txt", curr_dt, mail_id);
    strcpy(mail->rcpt, rcpt);
    strcpy(mail->from, from);
    strcpy(mail->reply, reply);
    trim_head_space(mail->reply);
    trim_tail_space(mail->reply);
    strcpy(mail->subject, subject);
    if(!(mail->body=strdup(body)))
    {
        free(mail);
        para_put_item(msg_out, SOAR_ERR_RESOURCE);
        goto RESPOND;
    }

    // 写邮件文件
    sprintf(mail_file, "%s/send/%s",
            share_info->_mail_.mail_store_dir, mail->file);
    if((fp=fopen(mail_file, "w")) == NULL)
    {
        errlog("%s: create mail_file(%s) fail", __func__, mail_file);
        free(mail->body);  free(mail);
        para_put_item(msg_out, SOAR_ERR_RESOURCE);
        goto RESPOND;
    }
    if(fputs(mail->from, fp) == EOF || fputs("\n", fp) == EOF ||
            fputs(mail->rcpt, fp) == EOF || fputs("\n", fp) == EOF ||
            fputs(mail->subject, fp) == EOF || fputs("\n", fp) == EOF ||
            fputs(mail->body, fp) == EOF)
    {
        errlog("%s: write mail_file(%s) fail", __func__, mail_file);
        fclose(fp);
        free(mail->body);  free(mail);
        para_put_item(msg_out, SOAR_ERR_RESOURCE);
        goto RESPOND;
    }
    fclose(fp);

    // 发送消息
    if(is_vip)
        sender_idx=SENDER_VIP_IDX;
    else
        sender_idx=random_int(1, share_info->sender_num-1, NULL);
    thrd_msgq_send(share_info->sender_msgq[sender_idx], &mail->link);
    para_put_item(msg_out, SOAR_ERR_OK);
    if(share_info->_errlog_.tracking)
        errlog("%s: worker(%d) send mail to sender(%d)",
                __func__, worker_idx, sender_idx); 

RESPOND:
    para_free(para);
    return(0);
}/*}}}*/

_SVC_FUNC(SEND)
{/*{{{*/
    return(_send(worker_ctl->worker_idx, msg, worker_ctl->msg_out,
                IS_FALSE));
}/*}}}*/

_SVC_FUNC(SENDVIP)
{/*{{{*/
    return(_send(worker_ctl->worker_idx, msg, worker_ctl->msg_out,
                IS_TRUE));
}/*}}}*/

static int _read(struct worker_ctl *worker_ctl,
        struct ep_active_slot *active_slot)
{/*{{{*/
    char *body;
    char *svc;
    int i;
    char *msg_out=worker_ctl->msg_out;

    active_slot->_u._read.msg[active_slot->_u._read.msg_size]=ZERO;

    if(active_slot->slot == worker_ctl->listener_slot)
    {
        static struct
        {
            char *svc;
            _P_SVC_FUNC p_svc_func;
        } a_svc_map[]=
        {
            { IN_SVC_NEW_SLOT, _SVC_FUNC_NEW_SLOT },
            { NULL, NULL }
        };

        body=active_slot->_u._read.msg;
        if(!(svc=para_get_item(active_slot->_u._read.msg, &body)) ||
                (i=strlen(svc)) <= 0 || i > NAME_SIZE)
        {
            errlog("%s: svc(%s) invalid, impossible",
                    __func__, (svc)?svc:SPACES);
            return(-1);
        }
        for(i=0; a_svc_map[i].svc; i++)
            if(strcmp(a_svc_map[i].svc, svc) == 0)
                return(a_svc_map[i].p_svc_func(worker_ctl, body));
        errlog("%s: svc(%s) not found, impossible", __func__, svc);
        return(-1);
    }
    else
    {
        static struct
        {
            char *svc;
            _P_SVC_FUNC p_svc_func;
        } a_svc_map[]=
        {
            { "SEND", _SVC_FUNC_SEND },
            { "SENDVIP", _SVC_FUNC_SENDVIP },
            { NULL, NULL }
        };
        int res;
        int msg_out_size;

        CLR_BUF(msg_out);
        body=active_slot->_u._read.msg;
        if(!(svc=para_get_item(active_slot->_u._read.msg, &body)) ||
                (i=strlen(svc)) <= 0 || i > NAME_SIZE)
        {
            errlog("%s: svc(%s) invalid",
                    __func__, (svc)?svc:SPACES);
            para_put_item(msg_out, SOAR_ERR_PARA);
            goto RESPOND;
        }
        for(i=0; a_svc_map[i].svc; i++)
            if(strcmp(a_svc_map[i].svc, svc) == 0)
                break;
        if(a_svc_map[i].svc)
        {   a_svc_map[i].p_svc_func(worker_ctl, body);   }
        else
        {
            errlog("%s: svc(%s) not found", __func__, svc);
            para_put_item(msg_out, SOAR_ERR_SVC_NOT_FOUND);
        }

RESPOND:
        msg_out_size=strlen(msg_out);
        if((res=ep_write(worker_ctl->ep, active_slot->slot,
                        msg_out, &msg_out_size)))
        {
            errlog("%s: worker(%d) ep_write fail, res:%d",
                    __func__, worker_ctl->worker_idx, res);
            return(-1);
        }
        return(0);
    }
}/*}}}*/

static int _abort(struct worker_ctl *worker_ctl,
                        struct ep_active_slot *active_slot)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;

    if(active_slot->slot == worker_ctl->listener_slot)
    {
        errlog("%s: worker(%d) revoke listener, res:%d, impossible",
                __func__, worker_idx, active_slot->_u._abort.error);
    }
    else
    {
        int slot_used_num;
        pthread_mutex_lock(&share_info->stat_lock);
        slot_used_num=--share_info->worker_stat[worker_idx].
            slot_used_num;
        pthread_mutex_unlock(&share_info->stat_lock);
        errlog("%s: worker(%d) slot_used:%d "
                "revoke slot(sock:%d, idx:%d), res:%d",
                __func__, worker_idx, slot_used_num,
                active_slot->_u._abort.sock,
                active_slot->slot_idx,
                active_slot->_u._abort.error);
    }
    return(0);
}/*}}}*/

void thrd_worker(struct worker_ctl *worker_ctl)
{/*{{{*/
    int res;
    struct ep_active_slot *active_list;

    while(1)
    {
        if((res=ep_poll(worker_ctl->ep, &active_list)) != 0)
        {
            errlog("%s: ep_poll fail, res:%d, impossible",
                    __func__, res);
            continue;
        }
        while(active_list)
        {
            switch(active_list->type)
            {
                case EP_SLOT_READ:
                    _read(worker_ctl, active_list);
                    break;
                case EP_SLOT_ABORT:
                    _abort(worker_ctl, active_list);
                    break;
                default:
                    errlog("%s: worker(%d) active slot type(%d) "
                            "invalid, impossible",
                            __func__, worker_ctl->worker_idx,
                            active_list->type);
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }
    }
}/*}}}*/
